package org.huntaway.hadoop;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class HadoopGrep {
	public   static   class  RegMapper  extends  MapReduceBase  implements  Mapper {

		   private  Pattern pattern;

		   public   void  configure(JobConf job) {
		   Pattern  =  Pattern.compile(job.get( " mapred.mapper.regex " ));
		  }

		   public   void  map(WritableComparable key, Writable value, OutputCollector output, Reporter reporter)
		     throws  IOException {
		   String text  =  ((Text) value).toString();
		   Matcher matcher  =  pattern.matcher(text);
		    if  (matcher.find()) {
		    output.collect(key, value);
		   }
		  }
		 }

		  private  HadoopGrep () {
		 }  //  singleton

		  public   static   void  main(String[] args)  throws  Exception {
		  
		  JobConf grepJob  =   new  JobConf(HadoopGrep. class );
		  grepJob.setJobName( " grep-search " );
		  grepJob.set( " mapred.mapper.regex " , args[ 2 ]);

		  grepJob.setInputPath( new  Path(args[ 0 ]));
		  grepJob.setOutputPath( new  Path(args[ 1 ]));
		  grepJob.setMapperClass(RegMapper. class );
		  grepJob.setReducerClass(IdentityReducer. class );
		      
		  JobClient.runJob(grepJob);
		 }

}
